package org.jie.guider.framework.springboot.kafka;

import java.util.List;
import java.util.concurrent.TimeUnit;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.jie.guider.framework.springboot.BlockThreadPool;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.listener.BatchAcknowledgingMessageListener;
import org.springframework.kafka.support.Acknowledgment;

/**
 * @author LiJie
 */
public class ChannelMessageListener<K, V> implements BatchAcknowledgingMessageListener<K, V> {
    private static final Logger LOG = LoggerFactory.getLogger(ChannelMessageListener.class);

    private Long channel;
    public static BlockThreadPool pool = new BlockThreadPool
        (5, "guider-shutdown", 30L, TimeUnit.SECONDS);

    public ChannelMessageListener(Long channel) {
        this.channel = channel;
    }
//
//    @Override
//    public void onMessage(ConsumerRecord<K, V> data, Acknowledgment acknowledgment) {
//        System.out.println(channel + " : " + data);
////        acknowledgment.acknowledge();
//    }

    @Override
    public void onMessage(List<ConsumerRecord<K, V>> data, Acknowledgment acknowledgment) {
        acknowledgment.acknowledge();
        for (ConsumerRecord<K, V> tmp : data) {
            pool.execute(new Runnable() {
                @Override
                public void run() {
                    try {
                        TimeUnit.SECONDS.sleep(35);
                    } catch (InterruptedException ignore) {
                    }
                    LOG.info("channel[{}] consume message : {}", channel, tmp);
                }
            });
        }
    }
}
